package org.dkd.config;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.RetryingBatchErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

import java.util.HashMap;
import java.util.Map;

/**
 * @className: KafkaConfig
 * @Auther: charlie
 * @Date: 2023/3/3 16:15
 * @Description:
 */
@Configuration
@Slf4j
public class KafkaConfig {

    @Autowired
    private KafkaProducerListener kafkaProducerListener;

    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaServers;

    @Value("${spring.kafka.producer.retries}")
    private Integer retries;

    @Value("${spring.kafka.producer.client-id}")
    private String clientId;

    @Value("${spring.kafka.producer.acks}")
    private String acks;

    @Value("${spring.kafka.producer.properties.linger.ms}")
    private Integer lingerMs;

    @Value("${spring.kafka.producer.properties.max.block.ms}")
    private Integer maxBlockMs;

    @Value("${spring.kafka.producer.properties.partitioner.class}")
    private String partitionerClass;

    @Bean("ProducerConfig")
    public Map<String, Object> producerConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, TextUtils.queryAndGetText(kafkaServers));
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.ACKS_CONFIG, acks);
        props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
       // props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, partitionerClass);
        return props;
    }

    @Bean(name = "producerFactory")
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfig());
    }

    /**
     * 自定义的测试环境kafka-template
     *
     * @return
     */
    @Primary
    @Bean(name = "kafkaTemplate")
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        kafkaTemplate.setProducerListener(kafkaProducerListener);
        return kafkaTemplate;
    }


    @Value("${spring.kafka.consumer.client-id}")
    private String consumerClientId;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private String enableAutoCommit;

    @Value("${spring.kafka.consumer.session-time-out}")
    private String sessionTimeOut;

    @Value("${spring.kafka.consumer.auto-offset-reset:earliest}")
    private String autoOffsetReset;

    @Value("${spring.kafka.consumer.max-poll-records}")
    private Integer maxPoolRecords;

    @Value("${spring.kafka.consumer.fetch-max-wait}")
    private Integer fetchMaxWait;

    @Value("${spring.kafka.consumer.fetch-min-size}")
    private String fetchMinSize;

    @Value("${spring.kafka.consumer.properties.max.poll.interval.ms}")
    private Integer maxPollIntervalMs;


    @Value("${spring.kafka.listener.poll-timeout}")
    private Integer pollTimeOut;


    @Bean("consumerConfig")
    public Map<String, Object> ConsumerConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerClientId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeOut);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPoolRecords);
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, fetchMaxWait);
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, fetchMinSize);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs);
        return props;
    }


    @Bean(name = "consumerFactory")
    public ConsumerFactory<String, String> ConsumerFactory() {
        DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(ConsumerConfig());
        consumerFactory.addListener(new ConsumerFactory.Listener<String, String>() {
            @Override
            public void consumerAdded(String id, Consumer<String, String> consumer) {
                log.info("consumer " + id + " add ........");
            }

            @Override
            public void consumerRemoved(String id, Consumer<String, String> consumer) {
                log.info("consumer " + id + " leave ........");
            }
        });
        return consumerFactory;
    }


    /**
     * @auther: charlie
     * @date: 2023/3/1 14:42
     * @desc: 监听工厂实例
     */
    @Primary
    @Bean(name = "defaultContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(ConsumerFactory());
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setConcurrency(1);
        factory.setMissingTopicsFatal(false);
        factory.getContainerProperties().setPollTimeout(pollTimeOut);
        factory.setBatchErrorHandler(new RetryingBatchErrorHandler(
                //5min 2次重试
                new FixedBackOff(5 * 60 * 1000, 2), (consumerRecord, e) -> {
            log.error("*******************************************消息处理异常:" + consumerRecord.value(), e);
        }
        ));
        return factory;
    }

}
